package com.kouen.kafka

import java.text.{DecimalFormat, SimpleDateFormat}
import javax.management.{AttributeNotFoundException, InstanceNotFoundException, MBeanServerConnection, ObjectName}
import javax.management.remote.{JMXConnector, JMXConnectorFactory, JMXServiceURL}

import org.apache.commons.codec.digest.DigestUtils

import scala.io.Source
import org.json4s._
import org.json4s.native.JsonMethods._
import redis.clients.jedis.{HostAndPort, Jedis, JedisCluster, JedisPool}
import java.util
import java.util.Date

import org.apache.commons.pool2.impl.GenericObjectPoolConfig


class Quota(conn: MBeanServerConnection, kafkaObj: KafkaObject, ipAndObjectName:String) {
  val name = ipAndObjectName

  @throws(classOf[java.io.IOException])
  @throws(classOf[java.net.SocketTimeoutException])
  def Send(
    url:String,
    connectTimeout: Int = 5000,
    readTimeout: Int = 5000,
    requestMethod: String = "GET") =
  {
    import java.net.{URL, HttpURLConnection}
    val trueUrl = new URL(url)
    val connection = trueUrl.openConnection.asInstanceOf[HttpURLConnection]
    connection.setConnectTimeout(connectTimeout)
    connection.setReadTimeout(readTimeout)
    connection.setRequestMethod(requestMethod)
    val inputStream = connection.getInputStream
    val content = io.Source.fromInputStream(inputStream).mkString
    if (inputStream != null) inputStream.close()
    content
  }

  def Update(): Map[String, Double] = {
    var ret: Map[String, Double] = Map()
    val df = new DecimalFormat("#.0000")
    for (attr <- kafkaObj.getAttribute){
      try{
        val value = conn.getAttribute(kafkaObj.getObjectName, attr)
        if (value != null)
          ret += (attr -> df.format(value.toString.toDouble).toDouble)
        else
          ret += (attr -> 0)
      }catch{
        case e: AttributeNotFoundException => None
        case e: InstanceNotFoundException => None
      }
    }
    ret
  }

  def PostData(server:Jedis, data:Map[String, Double])={
    val now:Date = new Date()
    val dateFormat:SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
    val timeFormat:SimpleDateFormat = new SimpleDateFormat("HH:mm:ss")
    val now_str = s"${dateFormat.format(now)}T${timeFormat.format(now)}";
    for {
      attr <- data.keys
      value = data(attr)
    }{
      val md5 = DigestUtils.md5Hex(s"$ipAndObjectName.$attr")
      server.select(0)
      server.publish("slow", s"$md5|$now_str@${String.valueOf(value)}")
//      server.set(md5, String.valueOf(value))
//      val regex = """.*(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3}).*""".r
//      val name = s"${ipAndObjectName match {case regex(_,_,_,ip4)=>ip4}}.${kafkaObj.ObjectType}"
//      server.select(1)
//      server.set(md5, String.valueOf(s"$name|$ipAndObjectName.$attr|number"))
    }
  }
}

object QuotaManager{
  var list = List[Quota]()
  var collectServer:String = ""
  var redisServer:String = ""
  def initAndCreate(file: String, kafkaObjectList: List[KafkaObject]): Unit ={
    val json = parse(file)
    implicit val formats = DefaultFormats

    (json \ "KafKa").extract[List[String]] foreach((ips) =>
    {
      val serviceURL = new JMXServiceURL(s"service:jmx:rmi:///jndi/rmi://$ips/jmxrmi")
      val connector = JMXConnectorFactory.connect(serviceURL, null)
      val conn = connector.getMBeanServerConnection()
      for (kfk <- kafkaObjectList) list :+= new Quota(conn, kfk, s"${kfk.getObjectName}.$ips")
    })

    collectServer = (json \ "CollectServer").extract[String]

    redisServer = (json \ "RedisServer").extract[String]

  }
  def Update()={
    val client = new JedisPool(
      new GenericObjectPoolConfig(),
      redisServer.split(":")(0),
      redisServer.split(":")(1).toInt,
      3000
    ).getResource()
    for(quota <- list) {
      val value = quota.Update()
      quota.PostData(client, value)
    }
  }
}

class KafkaObject(
  attr: List[String],
  template: String,
  key: String,
  name: String,
  tp: String,
  request: String="",
  topic: String="",
  partition: String = "")
{
  def ObjectType:String = name+" "+topic
  def getObjectName: ObjectName =
  {
    var objectName = template.replace("[key]", key).replace("[name]", name).replace("[type]", tp)
    if (request != "")
      objectName = objectName.replace("[request]", request)
    if (topic != "")
      objectName = objectName.replace("[topic]", topic)
    if (partition != "")
      objectName = objectName.replace("[partition]", partition)
    new ObjectName(objectName)
  }

  def getAttribute:List[String] = attr
}

object KafkaObject{
  var list = List[KafkaObject]()
  def initAndCreate(file: String) = {
    val json = parse(file)
    implicit val formats = DefaultFormats

    val topics = (json \ "topicAndPart").extract[Map[String, Int]]
    val monitor = (json \ "MonitorQuota").extract[Map[String, Map[String, Map[String, List[String]]]]]

    for {
      _key <- monitor.keys
      _keyValue = monitor(_key)
      _type <- _keyValue.keys
      _typeValue = _keyValue(_type)
      _name <- _typeValue("name")
      _attr = _typeValue("attr")
      _template = _typeValue("template").head
    } {
      val p_part = """\[partition\]""".r
      val p_topic = """\[topic\]""".r

      if (_typeValue.get("request").isDefined){
        _typeValue("request") foreach((_req)=> {
          if (p_part.findFirstIn(_template).isDefined){
            for {
              _t <- topics.keys
              _p = topics(_t)
            } {
              for(i <- 0 until _p) list :+= new KafkaObject(_attr, _template, _key, _name, _type, _req, _t, i.toString)
            }
          }else if (p_topic.findFirstIn(_template).isDefined){
            for {
              _t <- topics.keys
            } {
              list :+= new KafkaObject(_attr, _template, _key, _name, _type, _req, _t)
            }
          }
          else
            list :+= new KafkaObject(_attr, _template, _key, _name, _type, _req)
        })
      }else{
        if (p_part.findFirstIn(_template).isDefined){
          for {
            _t <- topics.keys
            _p = topics(_t)
          } {
            for(i <- 1 to _p) list :+= new KafkaObject(_attr, _template, _key, _name, _type, "", _t, i.toString)
          }
        }else if (p_topic.findFirstIn(_template).isDefined){
          for {
            _t <- topics.keys
          } {
            list :+= new KafkaObject(_attr, _template, _key, _name, _type, "", _t)
          }
        }
        else
          list :+= new KafkaObject(_attr, _template, _key, _name, _type, "")
      }

    }
  }
}

object CollectClient {
  def main(args: Array[String]) =
  {
    val lines = Source.fromResource("config.json").getLines().toList.mkString

    KafkaObject.initAndCreate(lines)
    QuotaManager.initAndCreate(lines, KafkaObject.list)
    QuotaManager.Update()
  }
}
